package server

import (
	"context"
	"fmt"
	"runtime"

	"git.julive.com/julive/user/client"
	"git.julive.com/julive/user/server/proto/user"
	"git.julive.com/julive/user/server/service"
	"git.julive.com/julive/user/server/subscriber"
	"git.julive.com/micro/pkg/broker"
	"git.julive.com/micro/pkg/grpcx"
	"git.julive.com/micro/pkg/logx"
	"git.julive.com/micro/pkg/tracing"
	gw_runtime "github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
	"github.com/opentracing/opentracing-go"
	"go.uber.org/zap"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"
	"google.golang.org/grpc/health"
	healthpb "google.golang.org/grpc/health/grpc_health_v1"
	"net"
	"net/http"
	_ "net/http/pprof"
	"sync"
)

var (
	GRPCPort int
	HTTPPort int
)

func Serve() (err error) {
	logger := logx.Logger.With(zap.String("service", "user"), zap.String("module", "server"))

	// 初始化上下文
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	// 初始化调用链追踪
	tracer, closer, err := tracing.InitTracer("julive-user-srv")
	if err != nil {
		return err
	}
	defer closer.Close()

	// 初始化kafka代理

	defer broker.Close()

	endpoint, err := broker.NewBroker("business")
	if err != nil {
		return err
	}
	// 设置消息队列事件处理器（可选）
	mqHandler := subscriber.NewUser(ctx)
	endpoint.Subscribe(
		broker.Processor{Topic: endpoint.Options().Topics["createuser"], Handler: mqHandler.CreateHandle, Retry: 1},
		broker.Processor{Topic: endpoint.Options().Topics["updateusercount"], Handler: mqHandler.UpdateCountHandle, Retry: 0},
	)

	cred := grpcx.NewGrpcCred()

	grpcServer := NewGrpcServer(cred.ServerCred(), tracer)

	listener, err := net.Listen("tcp", fmt.Sprintf(":%d", GRPCPort))
	if err != nil {
		return err
	}

	var wg sync.WaitGroup
	wg.Add(1)

	// 启动gGRC监听
	go func() {
		defer wg.Done()

		logger.Info("GrpcServer is running....")

		grpcServer.Serve(listener)
	}()

	defer client.Close()

	httpServer, err := NewHttpServer(ctx, cred.ClientCred())
	if err != nil {
		return err
	}

	go func() {
		defer wg.Done()

		logger.Info("HttpServer is running...")
		if err := httpServer.ListenAndServe(); err != nil {
			logger.Error("HttpServer running error", zap.Error(err))
		}
	}()

	// 退出时关闭gRPC、HTTP
	defer func() {
		grpcServer.GracefulStop()
		httpServer.Shutdown(ctx)
	}()

	wg.Wait()

	return nil
}

// 创建GrpcServer
// @cred 服务端证书 用于grpc双向认证
// @tracer 调用链追踪Tracer
func NewGrpcServer(cred credentials.TransportCredentials, tracer opentracing.Tracer) *grpc.Server {
	grpcServer := grpc.NewServer(grpc.Creds(cred), grpc.UnaryInterceptor(tracing.OpenTracingServerInterceptor(tracer)), grpc.StreamInterceptor(tracing.OpenTracingStreamServerInterceptor(tracer)))
	user.RegisterUserServiceServer(grpcServer, service.NewUserService())

	// 注册gRPC健康检测
	srv := health.NewServer()
	srv.SetServingStatus("", healthpb.HealthCheckResponse_SERVING)
	healthpb.RegisterHealthServer(grpcServer, srv)

	return grpcServer
}

// 创建GrpcServer
// @ctx 上下文
// @cred 服务端证书 用于grpc双向认证
func NewHttpServer(ctx context.Context, cred credentials.TransportCredentials) (*http.Server, error) {
	runtime.SetBlockProfileRate(0)

	localGrpc := fmt.Sprintf("localhost:%d", GRPCPort)

	router := http.DefaultServeMux
	// 健康检测接口，k8s调用
	probe := grpcx.NewHealth(localGrpc, cred)
	router.HandleFunc("/health", probe.Handler)

	// 注册gRPC-Gateway
	mux := gw_runtime.NewServeMux() //gw_runtime.WithMetadata(middleware.InitMeta))
	opts := []grpc.DialOption{grpc.WithTransportCredentials(cred)}
	err := user.RegisterUserServiceHandlerFromEndpoint(ctx, mux, localGrpc, opts)
	if err != nil {
		return nil, err
	}

	router.Handle("/", mux)

	httpServer := &http.Server{
		Addr:    fmt.Sprintf(":%d", HTTPPort),
		Handler: router,
	}

	return httpServer, nil
}
